﻿//Copyright © Steffen Zeidler 2010
using System;
using System.Collections.Generic;
using System.Linq.Expressions;
using System.Threading;
using System.Reactive.Subjects;
using System.Reactive.Linq;

namespace System.Linq
{
    public static class ObservableTrace
    {
        public struct TraceNotification
        {
            internal TraceNotification(long instance, string kind, object value)
            {
                this.instance = instance;
                this.kind = kind;
                this.current = value;
            }

            //each subscribe creates a new instance
            public long instance;
            //the name of the called interface method
            public string kind;
            //the value of the method parameter
            public object current;

            public override string ToString()
            {
                return string.Format("{0}: {1}({2})", instance, kind, current);
            }
        }

        /// <summary>
        /// Invokes the trace action on each call on the Rx interfaces IObservable, IObserver and IDisposable.
        /// </summary>
        /// <example>
        /// Observable.Range(0, 2).Trace(x => Console.WriteLine(x)).Repeat(2).Subscribe();
        /// 
        /// 1: Subscribe()
        /// 1: OnNext(0)
        /// 1: OnNext(1)
        /// 1: OnCompleted()
        /// 1: Dispose()
        /// 2: Subscribe()
        /// 2: OnNext(0)
        /// 2: OnNext(1)
        /// 2: OnCompleted()
        /// 2: Dispose()
        /// </example>
        public static IObservable<TSource> Trace<TSource>(this IObservable<TSource> source, Action<TraceNotification> trace)
        {
            long i = 0;
            return Observable.Create<TSource>(observer =>
            {
                long i1 = Interlocked.Increment(ref i);
                trace(new TraceNotification(i1, "Subscribe", ""));
                IDisposable disposable = source.Subscribe(
                    v => { trace(new TraceNotification(i1, "OnNext", v)); observer.OnNext(v); },
                    e => { trace(new TraceNotification(i1, "OnError", e)); observer.OnError(e); },
                    () => { trace(new TraceNotification(i1, "OnCompleted", "")); observer.OnCompleted(); });
                return () => { trace(new TraceNotification(i1, "Dispose", "")); disposable.Dispose(); };
            });
        }

        public struct TraceNotificationEx
        {
            //combinator index in expression
            public int idx;
            //x-axis, time
            public long x;
            //y-axis, instance, each subscribe creates a new instance
            public long y;
            public object combinator;
            public object expression;
            //the name of the called interface method
            public string method;
            //the value of the method parameter
            public object[] p;

            public override string ToString()
            {
                return string.Format("{0,9} {1,2}: {2}({3})", combinator, y, method, p.FirstOrDefault());
            }
        }

        /// <summary>
        /// Generates an observable trace sequence with values on each call on the Rx interfaces IObservable, IObserver and IDisposable.
        /// </summary>
        /// <example>
        /// ObservableTrace.Trace( () => Observable.Range(0, 2).Repeat(2) )
        /// .Subscribe(x => Console.WriteLine(x));
        /// 
        ///    Repeat  0: Subscribe()
        ///     Range  1: Subscribe()
        ///     Range  1: OnNext(0)
        ///    Repeat  0: OnNext(0)
        ///     Range  1: OnNext(1)
        ///    Repeat  0: OnNext(1)
        ///     Range  1: OnCompleted()
        ///     Range  2: Subscribe()
        ///     Range  2: OnNext(0)
        ///    Repeat  0: OnNext(0)
        ///     Range  2: OnNext(1)
        ///     Range  1: Dispose()
        ///    Repeat  0: OnNext(1)
        ///     Range  2: OnCompleted()
        ///     Range  2: Dispose()
        ///    Repeat  0: OnCompleted()
       /// </example>
        /// <param name="expression">Observable expression</param>
        /// <returns>Observable trace</returns>
        public static IObservable<ObservableTrace.TraceNotificationEx> Trace<T>(Expression<Func<IObservable<T>>> expression)
        {
            var expressionTrace = new ExpressionTrace();
            var expr2 = expressionTrace.Visit(expression);
            var action2 = (expr2 as Expression<Func<IObservable<T>>>).Compile();
            Observable.Start(() => expressionTrace.disposableExpression = action2().Finally(expressionTrace.subject.OnCompleted).Subscribe());
            return expressionTrace.subject;
        }

        //Special overload for the Marble Diagram Generator
        //Remark: the returned observable will not completed
        public static IObservable<ObservableTrace.TraceNotificationEx> Trace(Expression<Func<IDisposable>> expression)
        {
            var expressionTrace = new ExpressionTrace();
            var expr2 = expressionTrace.Visit(expression);
            var action2 = (expr2 as Expression<Func<IDisposable>>).Compile();
            Observable.Start(() => expressionTrace.disposableExpression = action2());
            return expressionTrace.subject;
        }

        //internal trace combinator
        private static IObservable<TSource> Trace<TSource>(this IObservable<TSource> source, ExpressionTrace state, int idx, MethodCallExpression expression)
        {
            return Observable.Create<TSource>(observer =>
            {
                long y1 = Interlocked.Increment(ref state.cy);
                Func<Delegate, object[], object> trace = (m, p) =>
                {
                    TraceNotificationEx rec = new TraceNotificationEx()
                    { 
                        x = Interlocked.Increment(ref state.cx), 
                        y = y1, 
                        combinator = expression.Method.Name, 
                        expression = expression,
                        method = m.Method.Name,
                        p = p, 
                        idx = idx
                    };
                    state.subject.OnNext(rec);
                    Interlocked.Increment(ref cInvoke);
                    var ret = m.DynamicInvoke(p);
                    Interlocked.Decrement(ref cInvoke);
                    return ret;
                };
                IDisposable disposable = (IDisposable)trace((Func<Action<TSource>, Action<Exception>, Action, IDisposable>)source.Subscribe,
                    new object[]{
                    (Action<TSource>)(v => trace((Action<TSource>)observer.OnNext, new object[] { v })),
                    (Action<Exception>)(e => trace((Action<Exception>)observer.OnError, new object[] { e })),
                    (Action)(() => trace((Action)observer.OnCompleted, empty))});
                return () => trace((Action)disposable.Dispose, empty);
            });
        }

        private static readonly object[] empty = new object[0];

        //Deadlock Detection
        private static int cInvoke;
        public static bool IsDeadlock()
        {
            return cInvoke > 0;
        }

        //internal expression visitor inserts trace combinator
        private class ExpressionTrace : ExpressionVisitor
        {
            internal void Reset()
            {
                if (disposableExpression != null)
                {
                    disposableExpression.Dispose();
                }
            }

            protected override Expression VisitMethodCall(MethodCallExpression node)
            {
                Type t = node.Method.ReturnType;
                if (t.IsGenericType && (t.GetGenericTypeDefinition() == traceT || (t = node.Method.ReturnType.GetInterface(traceT.Name)) != null))
                {
                    int idx1 = idx++;
                    return Expression.Call(
                        traceM.MakeGenericMethod(t.GetGenericArguments()[0]),
                        base.VisitMethodCall(node), Expression.Constant(this), Expression.Constant(idx1), Expression.Constant(node));
                }
                return base.VisitMethodCall(node);
            }

            private static readonly System.Reflection.MethodInfo traceM =
                ((Func<IObservable<int>, ExpressionTrace, int, MethodCallExpression, IObservable<int>>)ObservableTrace.Trace).Method.GetGenericMethodDefinition();
            private static readonly Type traceT = traceM.GetParameters()[0].ParameterType.GetGenericTypeDefinition();
            private int idx;
            internal IDisposable disposableExpression;
            internal long cx = -1, cy = -1;
            internal ISubject<TraceNotificationEx> subject = new ReplaySubject<TraceNotificationEx>(100);
        }
    }
}
